package com.wewins.fota.sys.service;


import com.alibaba.fastjson.JSON;
import com.wewins.fota.bean.bo.ExFotaLog;
import com.wewins.fota.bean.po.FotaLog;
import com.wewins.fota.common.constant.RedisConst;
import com.wewins.fota.service.FotaLogService;
import io.lettuce.core.RedisCommandTimeoutException;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.ListOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;

@Slf4j
@Service
public class RedisFotaLogService {
    @Setter
    @Value("${fota.db.storage-period:30s}")
    private Duration period;

    @Setter
    @Value("${fota.db.batch-count:500}")
    private int capacity;

    private Thread thread;

    private final RedisTemplate<String, Object> template;

    private final FotaLogService logService;

    // product:imei:srcV:operator
    private transient Map<String, ExFotaLog> batchMap = new HashMap<>();
    private transient long nextExecutor;
    private transient int count = 0;
    private transient String preLogStr;

    private static final Object object = new Object();

    @Autowired
    public RedisFotaLogService(RedisTemplate<String, Object> template, FotaLogService logService) {
        this.template = template;
        this.logService = logService;
    }

    @PostConstruct
    public void initData() {
        nextExecutor = System.currentTimeMillis() + period.toMillis();
    }

    public void initService() {
        Runnable runnable = () -> {
            log.info("start to init fota log writing thread.");
            consumeRedisLog();
        };
        thread = new Thread(runnable, "fota-log-by-redis");
        thread.start();
    }

    private void consumeRedisLog() {
        ListOperations<String, Object> ops = this.template.opsForList();
        Long size = 0L;
        while (true) {
            String logStr = "";
            try {
                size = ctlAndInsert(ops, size);
                logStr = (String) ops.rightPop(RedisConst.REDIS_KEY_LOG_LIST);
                consume(logStr);
                size--;
                count++;
            } catch (InterruptedException e) {
                log.warn("no log can consume, but sleep failed, restart log thread: ", e);
                this.initService();
                break;
            } catch (Throwable e) {
                if (!StringUtils.isEmpty(logStr) && !logStr.equals(preLogStr)) {
                    ops.rightPush(RedisConst.REDIS_KEY_LOG_LIST, logStr);
                    preLogStr = logStr;
                }
                if (e instanceof RedisCommandTimeoutException) {
                    log.error("redis command timeout, expected next time connected.");
                } else {
                    log.error("unknown exception happen, logStr:" + logStr + ", re-consume log:", e);
                }
            }
        }
    }

    private Long ctlAndInsert(ListOperations<String, Object> ops, Long size) throws InterruptedException {
        while (true) {
            insert2db();
            if (size == null || size <= 0) {
                size = ops.size(RedisConst.REDIS_KEY_LOG_LIST);
            }
            if (size == null || size <= 0) {
                Thread.sleep(5 * 1000);
                continue;
            }
            break;
        }
        return size;
    }

    private void insert2db() {
        if (batchMap.size() >= capacity || System.currentTimeMillis() >= nextExecutor) {
            if (batchMap.size() > 0) {
                log.info("this batch update fota log count:" + batchMap.size() + ", total:" + count);
                Map<String, ExFotaLog> tmp = batchMap;
                new Thread(() -> logService.batchInsertOrUpdate(tmp)).start();
                batchMap = new HashMap<>();
                count = 0;
            }
            nextExecutor = System.currentTimeMillis() + period.toMillis();
        }
    }

    private void consume(String logStr) {
        FotaLog fotaLog = JSON.parseObject(logStr, FotaLog.class);
        if (fotaLog != null) {
            ExFotaLog cacheExLog = batchMap.get(getBatchKey(fotaLog));
            if (cacheExLog != null) {
                FotaLog cacheLog = cacheExLog.getFotaLog();
                if (cacheLog != null) {
                    fotaLog.setId(cacheLog.getId());
                    fotaLog.setUpdateTime(fotaLog.getCreateTime());
                    fotaLog.setCreateTime(cacheLog.getCreateTime());
                    fotaLog.setCount(fotaLog.getCount() + cacheLog.getCount());
                    batchMap.put(getBatchKey(fotaLog), ExFotaLog.builder().fotaLog(fotaLog).dbType(cacheExLog.getDbType()).build());
                    return;
                }
            }

            if (logService.existLog(fotaLog)) {
                fotaLog.setUpdateTime(fotaLog.getCreateTime());
                batchMap.put(getBatchKey(fotaLog), ExFotaLog.builder().fotaLog(fotaLog).dbType(FotaLogService.TYPE_UPDATE).build());
                return;
            }

            batchMap.put(getBatchKey(fotaLog), ExFotaLog.builder().fotaLog(fotaLog).dbType(FotaLogService.TYPE_INSERT).build());
        }
    }

    private String getBatchKey(FotaLog fotaLog) {
        return fotaLog.getProduct() + ":" +
                fotaLog.getImei() + ":" +
                fotaLog.getFromV() + ":" +
                fotaLog.getOperator();
    }

    public void suspendThead() {
        synchronized (object) {
            try {
                this.thread.wait();
            } catch (InterruptedException e) {
                log.error("failed to wait writing fota log thread", e);
            }
        }
    }

    public void notifyThead() {
        synchronized (object) {
            this.thread.notify();
        }
    }
}
